/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.misc;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 异步信号量类
 * 该类提供了一种控制并发访问的机制，允许指定数量的线程同时访问共享资源
 * 与传统的信号量不同，异步信号量使用非阻塞方式来实现，这使得它在高并发场景下更加高效
 */
public class AsyncSemaphore {


    // 一个线程安全的计数器
    private final AtomicInteger counter;

    // 等待队列，泛型存了一个 CompletableFuture<Void>
    private final Queue<CompletableFuture<Void>> listeners = new ConcurrentLinkedQueue<>();

    // 相当于 init 方法，这里初始化计数器
    public AsyncSemaphore(int permits) {
        counter = new AtomicInteger(permits);
    }


    /**
     * 尝试获取资源，在指定的超时时间内等待完成
     */
    public boolean tryAcquire(long timeoutMillis) {
        CompletableFuture<Void> f = acquire();
        try {
            f.get(timeoutMillis, TimeUnit.MILLISECONDS);
            return true;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        } catch (ExecutionException e) {
            throw new IllegalStateException(e);
        } catch (TimeoutException e) {
            return false;
        }
    }


    /**
     * 获取监听器队列的大小
     */
    public int queueSize() {
        return listeners.size();
    }


    /**
     * 清空监听器队列
     */
    public void removeListeners() {
        listeners.clear();
    }


    /**
     *  acquire 方法的重载，区别是这里传一个 Runnable ，非阻塞直接执行，阻塞则等待后执行，是异步的
     *  主要实现是 CompletableFuture 的 thenAccept 方法，这里是一个链式调用
     */
    public void acquire(Runnable listener) {
        acquire().thenAccept(r -> listener.run());
    }


    /**
     * 获取一个CompletableFuture对象，用于异步获取资源
     */
    public CompletableFuture<Void> acquire() {
        // 这里首先创建了一个 CompletableFuture 加入到了等待队列中
        CompletableFuture<Void> future = new CompletableFuture<>();
        listeners.add(future);
        // 调用 tryRun() 方法，这里是限流的主要逻辑
        tryRun();
        return future;
    }



    /**
     * 限流主逻辑，尝试运行，注意这里是一个通用方法
     * 此方法利用一个计数器来跟踪剩余的任务数量，并在一个循环中不断尝试完成这些任务。
     * 如果没有剩余任务，则退出循环。
     * 使用CompletableFuture的complete方法来尝试完成任务，并在任务完成时调整计数器的值。
     */
    private void tryRun() {
        while (true) {
            // 先尝试 -1 如果计数器递减 >=0 则取队头的任务执行
            if (counter.decrementAndGet() >= 0) {
                // 从监听器队列中获取一个未完成的任务。
                CompletableFuture<Void> future = listeners.poll();
                // 等待队列没有任务, 计数器 +1 直接返回
                if (future == null) {
                    counter.incrementAndGet();
                    return;
                }
                // future.complete 将 future 置为完成，为后续 thenAccept 做铺垫
                // complete 里面调用了 postComplete() 该方法就是通知任务执行完成。触发后续依赖任务执行
                if (future.complete(null)) {
                    return;
                }
            }
            // 尝试运行失败后 +1 ，计数器已经 <=0 则直接返回 （后续的 tryRun 会在 release 里尝试调用）
            if (counter.incrementAndGet() <= 0) {
                return;
            }
        }
    }


    public int getCounter() {
        return counter.get();
    }

    public void release() {
        counter.incrementAndGet();
        tryRun();
    }

    @Override
    public String toString() {
        return "value:" + counter + ":queue:" + queueSize();
    }
    
    
    
}
